package com.informix.stream.cdc;

import com.gbasedbt.jdbc.IfxSmartBlob;
import com.informix.stream.impl.IfmxTableDescriptor;
import com.informix.stream.api.IfmxStreamEngine;
import com.informix.stream.api.IfmxStreamRecord;
import com.informix.stream.common.exception.IfxStreamException;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class IfxCDCEngine implements IfmxStreamEngine {
  public static class IfmxWatchedTable extends IfmxTableDescriptor {
    private static final AtomicInteger counter = new AtomicInteger(1);
    private int label = -1;
    private String[] columns;

    public IfmxWatchedTable(String databaseName, String namespace, String tableName) {
      super(databaseName, namespace, tableName);
      this.label = counter.getAndIncrement();
    }

    public IfmxWatchedTable(IfmxTableDescriptor desc) {
      super(desc.getDatabaseName(), desc.getNamespace(), desc.getTableName());
      this.label = counter.getAndIncrement();
    }

    public String getColumnDescriptorString() {
      return String.join(",", (CharSequence[]) this.columns);
    }

    public String[] getColumns() {
      return this.columns;
    }

    public IfmxWatchedTable columns(String[] columns) {
      this.columns = columns;
      return this;
    }

    public IfmxWatchedTable label(int label) {
      this.label = label;
      return this;
    }

    public int getLabel() {
      return this.label;
    }

    @Override
    public String toString() {
      return super.toString() + "::" + getColumnDescriptorString();
    }
  }

  public static class Builder {
    private final DataSource ds;
    private final List<IfmxWatchedTable> tables = new ArrayList<>();
    private int timeout = 5;
    private int buffer = 10240;
    private long sequencePosition = 0L;

    public Builder(DataSource ds) {
      this.ds = ds;
    }

    public Builder timeout(int timeout) {
      this.timeout = timeout;
      return this;
    }

    public Builder buffer(int bufferSize) {
      this.buffer = bufferSize;
      return this;
    }

    public int getBufferSize() {
      return this.buffer;
    }

    public Builder watchTable(String canonicalTableName, String... columns) {
      return watchTable(IfmxTableDescriptor.parse(canonicalTableName), columns);
    }

    public Builder watchTable(IfmxTableDescriptor desc, String... columns) {
      this.tables.add((new IfmxWatchedTable(desc)).columns(columns));
      return this;
    }

    public Builder watchTable(IfmxWatchedTable table) {
      this.tables.add(table);
      return this;
    }

    public int getTimeout() {
      return this.timeout;
    }

    public List<IfmxWatchedTable> getWatchedTables() {
      return this.tables;
    }

    public DataSource getDataSource() {
      return this.ds;
    }

    public Builder sequenceId(long position) {
      this.sequencePosition = position;
      return this;
    }

    public long getSequenceId() {
      return this.sequencePosition;
    }

    public IfxCDCEngine build() throws SQLException {
      return new IfxCDCEngine(this);
    }
  }

  private final Logger logger = LoggerFactory.getLogger(IfxCDCEngine.class);
  private final Builder builder;

  public static Builder builder(DataSource ds) {
    return new Builder(ds);
  }
  private IfxSmartBlob smartBlob;
  private final Connection con;
  private int sessionID;
  private final int bufferSize;
  private final byte[] buffer;
  private final IfxCDCRecordBuilder recordBuilder;
  private final int timeout;
  private final List<IfmxWatchedTable> capturedTables = new ArrayList<>();

  private boolean inlineLOB = false;

  private boolean isClosed = false;

  private boolean stopLoggingOnClose = true;

  private final long startingSequencePosition;

  private IfxCDCEngine(Builder builder) throws SQLException {
    this.builder = builder;
    this.con = builder.ds.getConnection();
    this.recordBuilder = new IfxCDCRecordBuilder(builder.ds.getConnection());
    this.timeout = builder.getTimeout();
    this.bufferSize = builder.getBufferSize();
    this.buffer = new byte[this.bufferSize];
    this.startingSequencePosition = builder.getSequenceId();
    this.capturedTables.addAll(builder.getWatchedTables());
  }

  @Override
  public void init() throws SQLException, IfxStreamException {
    String serverName;
    try (Statement s = this.con.createStatement();
        ResultSet rs =
            s.executeQuery(
                "SELECT env_value FROM sysmaster:sysenv where env_name = 'GBASEDBTSERVER'")) {
      rs.next();
      serverName = rs.getString(1).trim();
    }

    this.logger.debug("Server name detected: {}", serverName);
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_opensess(?,?,?,?,?,?)")) {

      cstmt.setString(1, serverName);

      cstmt.setInt(2, 0);
      cstmt.setInt(3, this.timeout);

      cstmt.setInt(4, 1);

      cstmt.setInt(5, 1);
      cstmt.setInt(6, 1);

      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        this.sessionID = rs.getInt(1);
        if (this.sessionID < 0) {
          throw new IfxStreamException(
              "Unable to create CDC session. Error code: " + this.sessionID);
        }
      }
    }
    this.smartBlob = new IfxSmartBlob(this.con);

    for (IfmxWatchedTable table : this.capturedTables) {
      watchTable(table);
    }
    activateSession();
  }

  @Override
  public IfmxStreamRecord getRecord() throws SQLException, IfxStreamException {
    int length = this.smartBlob.IfxLoRead(this.sessionID, this.buffer, this.buffer.length);
    if (length > -1) {
      IfmxStreamRecord r = this.recordBuilder.buildRecord(this.buffer);
      if (this.logger.isTraceEnabled()) {
        this.logger.trace("{}", r);
      }
      return r;
    }

    throw new IfxStreamException("IfxLoRead returned -1, no more data?");
  }

  private void watchTable(IfmxWatchedTable table) throws SQLException, IfxStreamException {
    this.logger.debug("Starting watch on table [{}]", table);
    enableFullRowLogging(table.getDesciptorString(), true);
    startCapture(table);
  }

  private void enableFullRowLogging(String tableName, boolean enable) throws IfxStreamException {
    this.logger.debug(
        "Setting full row logging on [{}] to '{}'", tableName, Boolean.valueOf(enable));
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_set_fullrowlogging(?,?)")) {
      cstmt.setString(1, tableName);
      if (enable) {
        cstmt.setInt(2, 1);
      } else {

        cstmt.setInt(2, 0);
      }
      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        int resultCode = rs.getInt(1);
        if (resultCode != 0) {
          throw new IfxStreamException("Unable to set full row logging. Error code: " + resultCode);
        }
      }
    } catch (SQLException ex) {
      throw new IfxStreamException("Unable to set full row logging ", ex);
    }
  }

  private void startCapture(IfmxWatchedTable table) throws SQLException {
    this.logger.debug("Starting capture on [{}]", table);
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_startcapture(?,?,?,?,?)")) {
      cstmt.setInt(1, this.sessionID);

      cstmt.setLong(2, 0L);
      cstmt.setString(3, table.getDesciptorString());
      cstmt.setString(4, table.getColumnDescriptorString());
      cstmt.setInt(5, table.getLabel());

      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        int resultCode = rs.getInt(1);
        if (resultCode != 0) {
          throw new SQLException(
              "CDCConnection: Unable to start cdc capture. Error code: " + resultCode);
        }
      }
    } catch (SQLException ex) {
      throw new SQLException("CDCConnection: Unable to start cdc capture ", ex);
    }
  }

  private void unwatchTable(IfmxWatchedTable table) throws IfxStreamException {
    endCapture(table);
    if (this.stopLoggingOnClose) {
      enableFullRowLogging(table.getDesciptorString(), false);
    }
  }

  private void endCapture(IfmxWatchedTable table) throws IfxStreamException {
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_endcapture(?,0,?)")) {
      cstmt.setInt(1, this.sessionID);
      cstmt.setString(2, table.getDesciptorString());

      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        int resultCode = rs.getInt(1);
        if (resultCode != 0) {
          throw new IfxStreamException("Unable to end cdc capture. Error code: " + resultCode);
        }
      }

    } catch (SQLException ex) {
      throw new IfxStreamException("Unable to end cdc capture ", ex);
    }
  }

  private void activateSession() throws IfxStreamException {
    this.logger.debug("Activating CDC session");
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_activatesess(?,?)")) {
      cstmt.setInt(1, this.sessionID);
      cstmt.setLong(2, this.startingSequencePosition);

      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        int resultCode = rs.getInt(1);
        if (resultCode != 0) {
          throw new IfxStreamException("Unable to activate session. Error code: " + resultCode);
        }
      }
    } catch (SQLException ex) {
      throw new IfxStreamException("Unable to activate session", ex);
    }
  }

  private void closeSession() throws IfxStreamException {
    this.logger.debug("Closing CDC session");
    try (CallableStatement cstmt =
        this.con.prepareCall("execute function gbasedbt.cdc_closesess(?)")) {
      cstmt.setInt(1, this.sessionID);

      try (ResultSet rs = cstmt.executeQuery()) {
        rs.next();
        int resultCode = rs.getInt(1);
        if (resultCode != 0) {
          throw new IfxStreamException("Unable to close session. Error code: " + resultCode);
        }
      }
    } catch (SQLException ex) {
      throw new IfxStreamException("Unable to close session", ex);
    }
  }

  @Override
  public void close() throws IfxStreamException {
    if (this.isClosed) {
      return;
    }
    this.logger.debug("Closing down CDC engine");
    IfxStreamException e = null;
    try {
      for (IfmxWatchedTable capturedTable : this.capturedTables) {
        unwatchTable(capturedTable);
      }
      closeSession();
    } catch (IfxStreamException ee) {
      e = ee;
    } finally {

      this.isClosed = true;
      try {
        this.con.close();
      } catch (SQLException ee) {
        IfxStreamException se = new IfxStreamException("Could not close main connection", ee);
        if (e == null) {
          e = se;
        } else {

          e.addSuppressed((Throwable) se);
        }
      }
      try {
        this.recordBuilder.close();
      } catch (SQLException ee) {
        IfxStreamException se = new IfxStreamException("Could not close record builder", ee);
        if (e == null) {
          e = se;
        } else {

          e.addSuppressed((Throwable) se);
        }
      }
    }
    if (e != null) {
      throw e;
    }
  }

  public boolean isInlineLOB() {
    return this.inlineLOB;
  }

  public Builder getBuilder() {
    return this.builder;
  }
}
